package com.qupeng.demo.kafka.kafkaspringbootstreams.common.kafka.streams;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.WallclockTimestampExtractor;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.KafkaStreamsConfiguration;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer;
import org.springframework.kafka.core.CleanupConfig;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
//@EnableKafkaStreams
public class KafkaStreamsConfig {
    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kStreamsConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-springboot-streams");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "172.26.143.96:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, CreatingTimestampExtractor.class.getName());
        return new KafkaStreamsConfiguration(props);
    }

    @Bean
    public CleanupConfig cleanupConfig() {
        return new CleanupConfig(true, true);
    }

    @Bean
    public StreamsBuilderFactoryBeanConfigurer configurer() {
        return fb -> fb.setStateListener((newState, oldState) -> {
            System.out.println("State transition from " + oldState + " to " + newState);
        });
    }

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME)
    public FactoryBean<StreamsBuilder> myKStreamBuilder(KafkaStreamsConfiguration kStreamsConfigs, CleanupConfig cleanupConfig) {
        StreamsBuilderFactoryBean myKStreamBuilderFactoryBean = new StreamsBuilderFactoryBean(kStreamsConfigs, cleanupConfig);
        myKStreamBuilderFactoryBean.setAutoStartup(true);
        myKStreamBuilderFactoryBean.setStateListener(new KafkaStreams.StateListener() {
            @Override
            public void onChange(KafkaStreams.State state, KafkaStreams.State state1) {
                System.out.println(state);
                System.out.println(state1);
            }
        });
        return myKStreamBuilderFactoryBean;
    }
}
